package drds.binlog.parse.inbound.mysql;

import com.lmax.disruptor.EventHandler;
import com.lmax.disruptor.LifecycleAware;
import drds.binlog.LogByteBlock;
import drds.binlog.LogContext;
import drds.binlog.LogDecoder;
import drds.binlog.LogEvent;
import drds.binlog.event.DeleteRowsLogEvent;
import drds.binlog.event.UpdateRowsLogEvent;
import drds.binlog.event.WriteRowsLogEvent;
import drds.binlog.parse.exception.BinlogParseException;
import drds.binlog.parse.inbound.TableMetaData;
import drds.binlog.protocol.entry_protocol_proto.Entry;


class SimpleParserStage implements EventHandler<MessageEvent>, LifecycleAware
{

    private MultiStageCoprocessor mysqlMultiStageCoprocessor;
    private LogDecoder decoder;
    private LogContext context;

    public SimpleParserStage(MultiStageCoprocessor mysqlMultiStageCoprocessor)
    {
        this.mysqlMultiStageCoprocessor = mysqlMultiStageCoprocessor;
        decoder = new LogDecoder(LogEvent.UNKNOWN_EVENT, LogEvent.ENUM_END_EVENT);
        context = new LogContext();
        if (mysqlMultiStageCoprocessor.gtidSet != null)
        {
            context.setGtidSet(mysqlMultiStageCoprocessor.gtidSet);
        }
    }

    public void onEvent(MessageEvent messageEvent, long sequence, boolean endOfBatch) throws Exception
    {
        try
        {
            LogEvent logEvent = messageEvent.getLogEvent();
            if (logEvent == null)
            {
                LogByteBlock logByteBlock = messageEvent.getLogByteBlock();
                logEvent = decoder.decode(logByteBlock, context);
                messageEvent.setLogEvent(logEvent);
            }

            int eventType = logEvent.getLogHeader().getType();
            TableMetaData tableMetaData = null;
            boolean needDmlParse = false;
            switch (eventType)
            {
                case LogEvent.WRITE_ROWS_EVENT_V1:
                case LogEvent.WRITE_ROWS_EVENT:
                    tableMetaData = mysqlMultiStageCoprocessor.logEventConvert.parseRowsEventForTableMeta((WriteRowsLogEvent) logEvent);
                    needDmlParse = true;
                    break;
                case LogEvent.UPDATE_ROWS_EVENT_V1:
                case LogEvent.UPDATE_ROWS_EVENT:
                    tableMetaData = mysqlMultiStageCoprocessor.logEventConvert.parseRowsEventForTableMeta((UpdateRowsLogEvent) logEvent);
                    needDmlParse = true;
                    break;
                case LogEvent.DELETE_ROWS_EVENT_V1:
                case LogEvent.DELETE_ROWS_EVENT:
                    tableMetaData = mysqlMultiStageCoprocessor.logEventConvert.parseRowsEventForTableMeta((DeleteRowsLogEvent) logEvent);
                    needDmlParse = true;
                    break;
                case LogEvent.ROWS_QUERY_LOG_EVENT:
                    needDmlParse = true;
                    break;
                default:
                    Entry entry = mysqlMultiStageCoprocessor.logEventConvert.parse(messageEvent.getLogEvent(), false);
                    messageEvent.setEntry(entry);
            }

            // 记录一下DML的表结构
            messageEvent.setNeedDmlParse(needDmlParse);
            messageEvent.setTableMetaData(tableMetaData);
        } catch (Throwable e)
        {
            mysqlMultiStageCoprocessor.binlogParseException = new BinlogParseException(e);
            throw mysqlMultiStageCoprocessor.binlogParseException;
        }
    }

    @Override
    public void onStart()
    {

    }

    @Override
    public void onShutdown()
    {

    }
}
